﻿//========================================================= 
/**@file  CommandAgent.h
 * @brief  
 * 
 * @date 2024-10-22 15:42:49 
 * @author yingbaizhou@tencent.com 
 * @version 1.0
 */ 
//--------------------------------------------------------- 
#ifndef _LIBZHOUYB_COMMANDAGENT_H_
#define _LIBZHOUYB_COMMANDAGENT_H_
//--------------------------------------------------------- 
#define INCLUDE_SOCKET_SOURCE
#define INCLUDE_ASYNC_SOURCE

#include "CommandDriver.h"

#include "../../wrapper/rapidjson/rapidjson_extractor.h"
using namespace zhou_yb::wrapper::rapidjson_extractor;

#include "../tools/Runable.h"
using zhou_yb::application::tools::Runable;
//---------------------------------------------------------
#define LOG LogCat_Default(__FUNCTION__)
//--------------------------------------------------------- 
namespace zhou_yb {
namespace application {
namespace driver {
//--------------------------------------------------------- 
struct AgentTrans
{
    uint id;
    Timer::TickType timestamp;
};
//--------------------------------------------------------- 
enum CommandType
{
    enum_CmdUnknown = 0,
    enum_ServerCall,
    enum_RemoteCall
};
enum MessageType
{
    enum_MsgUnknown = 0,
    enum_MsgReq,
    enum_MsgRes,
    enum_MsgNotify
};
//--------------------------------------------------------- 
struct IAgentNotifier
{
    virtual void Notify(const ByteArray& cmd, IStringArguments* arg) = 0;
};
//--------------------------------------------------------- 
struct ICommandAgent : public CommandCollection
{
    Ref<IAgentNotifier> observer;

    virtual bool Update(Timer::TickType tick) = 0;
    virtual bool FixedUpdate(Timer::TickType tick, uint delta) = 0;
};
//--------------------------------------------------------- 
struct CommandAgentHepler
{
    //----------------------------------------------------- 
    template<class T>
    static T config_option(IniGroup& grp, const char* key, const T& default_val = T())
    {
        IniGroup::const_iterator itr = grp.Get(key);
        if (itr == grp.end())
        {
            return default_val;
        }
        T val;
        if (!ArgConvert::FromString<T>(itr->Value, val))
            return default_val;
        return val;
    }
    //----------------------------------------------------- 
    static bool execute(CommandCollection& api, const ByteArray& cmd, IStringArguments& arg, IStringArguments& rlt)
    {
        Ref<ComplexCommand> command = api.LookUp(cmd);
        if (command.IsNull())
        {
            string msg = "<";
            msg += cmd.GetString();
            msg += "> is not exists";
            rlt.Put("Msg", msg);
            return false;
        }
        return command->OnCommand(arg, rlt);
    }
    //----------------------------------------------------- 
    /**
     * @brief 初始化配置
     * @date 2021-02-26 20:30
     */
    static bool auto_init(IniFile& cfg, const char* init_grp, CommandCollection& api)
    {
        StringArguments arg;
        StringArguments rlt;
        IniGroup& grp = cfg[_strput(init_grp)];

        for (IniGroup::iterator itr = grp.begin(); itr != grp.end(); ++itr)
        {
            if (itr->Key.length() < 1)
                continue;

            ByteArray cmd(itr->Key.c_str(), itr->Key.length());

            arg.Clear();
            rlt.Clear();

            if (itr->Value.length() > 0)
            {
                IniGroup& param_grp = cfg[itr->Value];
                for (IniGroup::iterator param_itr = param_grp.begin(); param_itr != param_grp.end(); ++param_itr)
                {
                    arg.Put(param_itr->Key, param_itr->Value);
                }
            }
            Ref<ComplexCommand> command = api.LookUp(cmd);
            if (!command.IsNull())
            {
                command->OnCommand(arg, rlt);
            }
        }

        return true;
    }
    //----------------------------------------------------- 
};
//--------------------------------------------------------- 
class HostAgent : public Runable, public InterruptBehavior, public IAgentNotifier, public RefObject
{
public:
    //----------------------------------------------------- 
    struct NotifyMessage
    {
        ByteBuilder cmd;
        uint from;
        ByteBuilder data;
        Timer::TickType expire_time;
    };
    //----------------------------------------------------- 
    /**
     * @brief 打包指定的remote_call调用参数
     * @date 2019-03-14 16:02
     */
    static size_t PackCMD(const ByteArray& command, uint id, CommandType cmdType, MessageType msgType,
        uint from, uint to, const ByteArray& sJson, ByteBuilder& cmd)
    {
        size_t lastLen = cmd.GetLength();
        DevCommand::FromAscii("01 A1 00000000", cmd);

        size_t cmdlen = command.GetLength() + 1;
        size_t datalen = sJson.GetLength() + 1;
        size_t bodylen = 4 + cmdlen + 4 + (1 + 1 + 4 + 4) + 4 + datalen;
        // 4=命令头长度 4=ID 4=DATA长度
        size_t len = cmd.GetLength() + 4 + bodylen;
        ByteConvert::FromObject<size_t>(len, cmd, 4);

        // CMD
        ByteConvert::FromObject<size_t>(cmdlen, cmd, 4);
        cmd += command;
        cmd += static_cast<byte>(0x00);
        // ID
        ByteConvert::FromObject<size_t>(id, cmd, 4);
        // CommandType
        cmd += static_cast<byte>(cmdType);
        // MessageType
        cmd += static_cast<byte>(msgType);
        // From
        ByteConvert::FromObject<uint>(from, cmd, 4);
        // To
        ByteConvert::FromObject<uint>(to, cmd, 4);
        // DATA
        ByteConvert::FromObject<size_t>(datalen, cmd, 4);
        cmd += sJson;
        cmd += static_cast<byte>(0x00);
        return (cmd.GetLength() - lastLen);
    }
    static bool IsFormat(const ByteArray& data, byte& bCmd, size_t& len)
    {
        if (data.GetLength() < 10)
            return false;
        size_t bodylen = 0;
        ByteConvert::ToObject<size_t>(data.SubArray(6, 4), bodylen);
        if (bodylen > data.GetLength())
            return false;
        bCmd = data[1];
        len = bodylen;
        return true;
    }
    static bool UnpackCMD(const ByteArray& data, ByteArray& cmd, uint& id, CommandType& cmdType,
        MessageType& msgType, uint& from, uint& to, ByteArray& rJson)
    {
        byte bCmd = 0x00;
        size_t bodylen = 0;
        if (!IsFormat(data, bCmd, bodylen) || bCmd != 0xA1)
            return false;

        size_t cmdlen = 0;
        ByteArray body = data.SubArray(6 + 4);

        // Command
        uint offset = 0;
        uint sublen = 4;
        ByteConvert::ToObject<size_t>(body.SubArray(offset, sublen), cmdlen);
        offset += sublen;
        cmd = body.SubArray(offset, cmdlen);
        offset += cmdlen;

        // ID
        sublen = 4;
        ByteConvert::ToObject<uint>(body.SubArray(offset, sublen), id);
        offset += sublen;
        // CmdType
        sublen = 1;
        uint tmp = 0;
        ByteConvert::ToObject<uint>(body.SubArray(offset, sublen), tmp);
        cmdType = static_cast<CommandType>(tmp);
        offset += sublen;
        // MsgType
        sublen = 1;
        ByteConvert::ToObject<uint>(body.SubArray(offset, sublen), tmp);
        msgType = static_cast<MessageType>(tmp);
        offset += sublen;
        // From
        sublen = 4;
        ByteConvert::ToObject<uint>(body.SubArray(offset, sublen), from);
        offset += sublen;
        // To
        sublen = 4;
        ByteConvert::ToObject<uint>(body.SubArray(offset, sublen), to);
        offset += sublen;
        // BodyLen
        sublen = 4;
        ByteConvert::ToObject<size_t>(body.SubArray(offset, sublen), bodylen);
        offset += sublen;
        // Body
        rJson = body.SubArray(offset, bodylen);
        return true;
    }
    static bool RecvCMD(IInteractiveTrans& dev, ByteBuilder& data)
    {
        const size_t MIN_LEN = 10;
        if (!DevHelper::IsHelperSuccess(IInteractiveTransHelper::ReadAssert(dev, MIN_LEN, data)))
            return false;
        size_t len = 0;
        ByteConvert::ToObject(data.SubArray(6, 4), len);
        return DevHelper::IsHelperSuccess(IInteractiveTransHelper::ReadAssert(dev, len, data));
    }
    //----------------------------------------------------- 
public:
    //----------------------------------------------------- 
    /// 心跳间隔
    uint HeartBeatIntervalMs;
    /// 事务超时时间
    uint TransTimeoutMs;
    /// 通知过期时间
    uint NotifyTimeoutMs;
    /// 空闲等待间隔
    uint IdleIntervalMs;
    //----------------------------------------------------- 
protected:
    //----------------------------------------------------- 
    Ref<ICommandAgent> _api;

    ByteBuilder _cachedBuff;
    AsyncSocketDevice _dev;
    uint _tid;
    uint _idx;
    uint _req_idx;
    RapidJsonArgParser _parser;

    Timer::TickType _heart_beat_time_tick;
    uint _heart_beat_last_tid;
    uint _wait_duration;

    ByteBuilder _cmd;
    ByteBuilder _data;
    StringArguments _arg;
    StringArguments _rlt;

    std::list<AgentTrans> _trans_pool;
    std::list<NotifyMessage> _notify_pool;
    //----------------------------------------------------- 
    bool _Wait(ByteBuilder& command, ByteBuilder& data, uint& id, uint& from, uint& to, CommandType& cmdType, MessageType& msgType)
    {
        byte bCmd = 0x00;
        size_t len = 0;
        // 没有历史数据,触发读
        if (!IsFormat(_cachedBuff, bCmd, len))
        {
            Timer::TickType t = Timer::TimeTick();
            if (!RecvCMD(_dev, _cachedBuff))
                return false;
            if (!IsFormat(_cachedBuff, bCmd, len))
                return false;
        }
        if (bCmd == 0xA1)
        {
            ByteArray cmd;
            ByteArray sJson;
            UnpackCMD(_cachedBuff, cmd, id, cmdType, msgType, from, to, sJson);
            if (cmd[cmd.GetLength() - 1] == 0)
                cmd = cmd.SubArray(0, cmd.GetLength() - 1);
            if (sJson[sJson.GetLength() - 1] == 0)
                sJson = sJson.SubArray(0, sJson.GetLength() - 1);
            LOG.debug() << "recv " << cmd.GetString() << " id:" << id << ' ' << from << "->" << to << ' ' << sJson.GetString();
            command.Append(cmd);
            data.Append(sJson);
        }
        else
        {
            LOG.warn() << "recv error cmd : " << _hex_num(bCmd);
        }
        _cachedBuff.RemoveFront(len);
        return true;
    }
    void _ProcMsg(Timer::TickType tick, ByteBuilder& command, ByteBuilder& data, uint tid, uint from, uint to, CommandType cmdType, MessageType msgType)
    {
        bool bOK = false;
        switch (msgType)
        {
        case enum_MsgReq:
            _req_idx = from;

            _arg.Clear();
            _rlt.Clear();

            LOG.info() << "on call : " << _cmd.GetString() << " \"" << _data.GetString() << "\"";
            if (_data.GetLength() > 0)
            {
                _parser.Parse(_data, _arg);
            }
            bOK = _HandleReq(_cmd, _arg, _rlt);
            StringArguments::Set(_rlt, "IsOK", ArgConvert::ToString(bOK));
            _SendNotify(_cmd, &_rlt);
            break;
        case enum_MsgRes:
            for (std::list<AgentTrans>::iterator itr = _trans_pool.begin(); itr != _trans_pool.end(); ++itr)
            {
                if (itr->id == tid)
                {
                    _trans_pool.erase(itr);
                    break;
                }
            }
            if (StringConvert::Compare(_cmd, "agent_svr_heart_beat"))
            {
                _idx = to;
            }
            break;
        case enum_MsgNotify:
            _notify_pool.push_back(NotifyMessage());
            {
                NotifyMessage& msg = _notify_pool.back();
                msg.cmd = _cmd;
                msg.data = _data;
                msg.from = from;
                msg.expire_time = tick + NotifyTimeoutMs;
            }
            break;
        default:
            LOG.error() << "recv error msg_type : " << static_cast<int>(msgType);
            break;
        }
    }
    bool _Send(const ByteArray& cmd, CommandType cmdType, MessageType msgType = enum_MsgReq, uint to = 0, const IStringArguments* arg = NULL)
    {
        ByteBuilder send(64);
        ByteBuilder param(64);

        if (arg != NULL)
        {
            _parser.ToString(*arg, param);
        }
        if (msgType == enum_MsgNotify)
        {
            LOG.info() << "notify :" << cmd.GetString() << " \"" << param.GetString() << "\"";
        }
        LOG.debug() << "send " << cmd.GetString() << " id:" << (_tid + 1) << ' ' << _idx << "->" << to << ' ' << param.GetString();

        PackCMD(cmd, _tid + 1, cmdType, msgType, _idx, to, param, send);
        if (!_dev.Write(send))
            return false;
        AgentTrans t;
        _tid += 1;
        t.id = _tid;
        t.timestamp = Timer::TimeTick() + TransTimeoutMs;

        _trans_pool.push_back(t);
        return true;
    }
    bool _SendNotify(const ByteArray& cmd, IStringArguments* arg = NULL)
    {
        return _Send(cmd, enum_RemoteCall, enum_MsgNotify, _req_idx, arg);
    }
    bool _HandleReq(const ByteArray& cmd, IStringArguments& arg, IStringArguments& rlt)
    {
        if (StringConvert::Compare(cmd, "Exit"))
        {
            _is_running = false;
            return true;
        }
        return CommandAgentHepler::execute(_api, cmd, arg, rlt);
    }
    void _UpdateTrans(Timer::TickType tick)
    {
        /* 处理事务超时 */
        for (std::list<AgentTrans>::iterator itr = _trans_pool.begin(); itr != _trans_pool.end();)
        {
            if (itr->timestamp < tick)
            {
                LOG.error() << "trans is timeout : " << itr->id;
                if (itr->id == _heart_beat_last_tid)
                {
                    LOG.error() << "agent server maybe is busy.";
                    Shutdown();
                    break;
                }

                itr = _trans_pool.erase(itr);
            }
            else
            {
                ++itr;
            }
        }
        /* 处理过期的通知 */
        for (std::list<NotifyMessage>::iterator itr = _notify_pool.begin(); itr != _notify_pool.end();)
        {
            if (itr->expire_time < tick)
            {
                LOG.error() << "notify is timeout : " << itr->cmd.GetString();
                itr = _notify_pool.erase(itr);
            }
            else
            {
                ++itr;
            }
        }
    }
    bool _UpdateMsg(Timer::TickType tick)
    {
        uint tid = 0;
        uint from = 0;
        uint to = 0;
        CommandType cmdType = enum_CmdUnknown;
        MessageType msgType = enum_MsgUnknown;
        bool bOK = false;

        _cmd.Clear();
        _data.Clear();

        if (!_Wait(_cmd, _data, tid, from, to, cmdType, msgType))
        {
            // 当前有通信数据时不触发心跳
            if (tick > _heart_beat_time_tick)
            {
                if (!_Send("agent_svr_heart_beat", enum_ServerCall))
                {
                    LOG.error() << "send message to agent server failed, server maybe is closed.";
                    return false;
                }
                _heart_beat_last_tid = _tid;
                _heart_beat_time_tick = tick + HeartBeatIntervalMs;
                return true;
            }
            return false;
        }
        // Agent客户端只处理 RemoteCall
        if (cmdType != enum_RemoteCall && msgType == enum_MsgReq)
        {
            LOG.error() << "recv error cmd_type : " << static_cast<int>(cmdType);
            return true;
        }
        _ProcMsg(tick, _cmd, _data, tid, from, to, cmdType, msgType);
        return true;
    }
    bool _Request(const ByteArray& command, CommandType call, uint req_to, uint timeoutMs, const IStringArguments* arg, IStringArguments* rlt)
    {
        if (!_Send(command, call, enum_MsgReq, req_to, arg))
            return false;

        ByteBuilder cmd(16);
        ByteBuilder data(128);

        uint req_tid = _tid;
        uint tid = 0;
        uint from = 0;
        uint to = 0;
        CommandType cmdType = enum_CmdUnknown;
        MessageType msgType = enum_MsgUnknown;

        Timer t;
        while (t.Elapsed() < timeoutMs)
        {
            if (_Wait(cmd, data, tid, from, to, cmdType, msgType))
            {
                if (StringConvert::Compare(cmd, command))
                {
                    for (std::list<AgentTrans>::iterator itr = _trans_pool.begin(); itr != _trans_pool.end(); ++itr)
                    {
                        if (itr->id == req_tid)
                        {
                            _trans_pool.erase(itr);
                            break;
                        }
                    }
                    if (_idx == 0)
                    {
                        _idx = to;
                    }
                    if (rlt)
                    {
                        _parser.Parse(data, *rlt);
                    }
                    return true;
                }
            }
        }
        return false;
    }
    //----------------------------------------------------- 
public:
    //----------------------------------------------------- 
    /**
     * @brief 以守护进程的方式运行
     *
     * @param [in] url : 连接的服务器地址 127.0.0.1:9999
     * @param [in] cache : 配置文件及日志输出路径
     * @param [in] agent_ini : 配置文件名(不含扩展名)
     *
     * @date 2021-02-20 16:50
     */
    static bool domain(ICommandAgent& api, const char* url, const char* cache, const char* agent_ini)
    {
        IniFile ini;
        string config_path = _strput(cache);
        if (!config_path.empty())
        {
            config_path += PATH_SEPARATOR;
        }
        config_path += (_is_empty_or_null(agent_ini) ? "agent" : agent_ini);
        config_path += ".ini";

        string logdir = _strput(cache);
        if (logdir.length() > 0)
        {
            logdir += PATH_SEPARATOR;
        }
        logdir += "log";
        LogCatHelper::DefaultWithFolder(logdir.c_str(), "Agent");

        if (!ini.Open(config_path.c_str()))
        {
            LOG.warn() << "can't found config file : " << config_path;
        }
        IniGroup& grp = ini["agent"];
        string ip = "";
        ushort port = 0;
        ByteBuilder url_data(url);
        size_t index = StringConvert::IndexOf(url_data, ':');
        if (index != SIZE_EOF)
        {
            port = ArgConvert::FromString<ushort>(url_data.SubArray(index + 1).GetString());
            url_data[index] = 0;
            ip = url_data.GetString();
        }
        else
        {
            ip = CommandAgentHepler::config_option<string>(grp, "IP", "");
            port = CommandAgentHepler::config_option<ushort>(grp, "Port", 0);
        }
        uint timeoutMs = CommandAgentHepler::config_option<uint>(grp, "TimeoutMs", 10000);
        string sLogLv = CommandAgentHepler::config_option<string>(grp, "LogLv", "INFO");
        LogCatHelper::Default().SetPriority("Console", LogCat::ConvertFrom(sLogLv.c_str()));

        if (ip.length() < 1 || port == 0)
        {
            LOG.error() << "IP or Port is error (" << ip << ':' << port << ")";
            return false;
        }

        HostAgent agent;
        agent.Serve(api);
        bool is_auto_init = CommandAgentHepler::config_option<bool>(grp, "IsAutoInit", false);
        if (is_auto_init && !CommandAgentHepler::auto_init(ini, "auto_init", api))
        {
            LOG.error() << "Agent init failed.";
            return false;
        }

        string name = CommandAgentHepler::config_option<string>(grp, "Name", "none");
        uint max_retry_count = CommandAgentHepler::config_option<uint>(grp, "RetryCount", 3);

        agent.TransTimeoutMs = CommandAgentHepler::config_option<uint>(grp, "TransTimeoutMs", 3000);
        agent.IdleIntervalMs = CommandAgentHepler::config_option<uint>(grp, "IdleIntervalMs", 200);
        agent.HeartBeatIntervalMs = CommandAgentHepler::config_option<uint>(grp, "HeartBeatIntervalMs", 3000);
        agent.FixedUpdateIntervalMs = CommandAgentHepler::config_option<uint>(grp, "UpdateIntervalMs", 20);
        agent.IsAllowUpdateSleep = CommandAgentHepler::config_option<bool>(grp, "IsAllowUpdateSleep", true);

        uint retry_count = 0;
        for (retry_count = 0; retry_count < max_retry_count; ++retry_count)
        {
            if (agent.Connect(ip.c_str(), port, timeoutMs, name.c_str()))
            {
                retry_count = 0;
                LOG.info() << "connect to agent server success.";
                break;
            }
            LOG.error() << "connect to agent server timeout. retry count : " << retry_count + 1;
            Timer::Wait(agent.TransTimeoutMs);
        }
        // 成功连接
        if (retry_count < max_retry_count)
        {
            LOG.info() << "Start run agent...";
            uint fps = agent.Run();
            LOG.info() << "FPS:" << fps << "ms";
        }
        LOG.warn() << "exit to domain.";
        return true;
    }
    //----------------------------------------------------- 
public:
    //----------------------------------------------------- 
    HostAgent() : Runable()
    {
        _tid = 0;
        _idx = 0;
        _req_idx = 0;

        _heart_beat_time_tick = 0;
        _heart_beat_last_tid = 0;
        _wait_duration = 0;

        _is_running = false;

        HeartBeatIntervalMs = 3000;
        TransTimeoutMs = 3000;
        NotifyTimeoutMs = 10000;
        IdleIntervalMs = 100;
    }
    //----------------------------------------------------- 
    virtual bool Update(Timer::TickType tick) { return _api->Update(tick); }
    /// 更新socket
    virtual bool FixedUpdate(Timer::TickType tick, uint delta)
    {
        bool is_busy = _api->FixedUpdate(tick, delta);
        bool is_msg = _UpdateMsg(tick);
        _wait_duration += delta;
        if (_wait_duration >= IdleIntervalMs)
        {
            _wait_duration = 0;
            // 中断支持
            if (InterruptBehavior::Implement(*this))
            {
                Shutdown();
                return true;
            }
            _UpdateTrans(tick);
        }
        return is_busy || is_msg;
    }
    virtual void Notify(const ByteArray& cmd, IStringArguments* arg)
    {
        _SendNotify(cmd, arg);
    }
    //----------------------------------------------------- 
    void Serve(Ref<ICommandAgent> api)
    {
        _api = api;
        _api->observer = *this;
    }
    /// 连接服务端
    bool Connect(const char* ip, ushort port, uint timeoutMs, const char* name)
    {
        LOG.info() << _expression(ip) << ' ' << _expression(port) << ' ' << _expression(timeoutMs);
        const uint connect_timeoutMs = 1000;
        _dev.SetWaitTimeout(connect_timeoutMs);
        _dev.Interrupter = Interrupter;

        // Connect to Server
        Timer timer;
        while (timer.Elapsed() < timeoutMs)
        {
            if (DevHelper::IsHelperSuccess(SocketDeviceHelper::Connect(_dev, ip, port)))
            {
                LOG.info() << "Connect to Server Success.";
                if (Heartbeat(name, timeoutMs))
                {
                    LOG.info() << "Heartbeat success, self:" << _idx;
                    _dev.SetWaitTimeout(DEV_OPERATOR_INTERVAL);
                    _dev.SetOperatorInterval(5);
                    return true;
                }
                _dev.Close();
                LOG.error() << "send Heartbeat to Server failed.";
                return false;
            }
            else
            {
                LOG.warn() << "Connect agent server failed.";
            }
            Timer t;
            while (t.Elapsed() < connect_timeoutMs)
            {
                /* 中断支持 */
                if (InterruptBehavior::Implement(*this))
                {
                    LOG.error() << "Connecting is Canceled.";
                    return false;
                }
                Timer::Wait(100);
            }
        }
        LOG.error() << "Connect to Server Timeout.";
        return false;
    }
    /// 发出请求
    bool Request(const ByteArray& command, uint req_to, uint timeoutMs, const IStringArguments* arg = NULL, IStringArguments* rlt = NULL)
    {
        if (!_Send(command, enum_RemoteCall, enum_MsgReq, req_to, arg))
            return false;

        ByteBuilder cmd(16);
        ByteBuilder data(128);

        uint tid = 0;
        uint from = 0;
        uint to = 0;
        CommandType cmdType = enum_CmdUnknown;
        MessageType msgType = enum_MsgUnknown;

        Timer t;
        while (t.Elapsed() < timeoutMs)
        {
            cmd.Clear();
            data.Clear();
            if (_Wait(cmd, data, tid, from, to, cmdType, msgType))
            {
                _ProcMsg(Timer::TimeTick(), cmd, data, tid, from, to, cmdType, msgType);
                if (StringConvert::Compare(cmd, command))
                {
                    if (rlt)
                    {
                        _parser.Parse(data, *rlt);
                    }
                    return true;
                }
            }
        }
        return false;
    }
    /// 等待通知
    bool Expect(const ByteArray& notify, uint notify_from, uint timeoutMs, IStringArguments* rlt)
    {
        for (std::list<NotifyMessage>::iterator itr = _notify_pool.begin();itr != _notify_pool.end(); ++itr)
        {
            if (StringConvert::Compare(itr->cmd, notify) && itr->from == notify_from)
            {
                if (rlt)
                {
                    _parser.Parse(itr->data, *rlt);
                }
                _notify_pool.erase(itr);
                return true;
            }
        }

        ByteBuilder cmd(16);
        ByteBuilder data(128);

        uint tid = 0;
        uint from = 0;
        uint to = 0;
        CommandType cmdType = enum_CmdUnknown;
        MessageType msgType = enum_MsgUnknown;

        Timer t;
        while (t.Elapsed() < timeoutMs)
        {
            cmd.Clear();
            data.Clear();
            if (_Wait(cmd, data, tid, from, to, cmdType, msgType))
            {
                if (msgType == enum_MsgNotify && from == notify_from && StringConvert::Compare(cmd, notify))
                {
                    if (rlt)
                    {
                        _parser.Parse(data, *rlt);
                    }
                    return true;
                }
            }
        }
        return false;
    }
    /// 发出心跳请求
    bool Heartbeat(const char* name, uint timeoutMs)
    {
        StringArguments arg;
        StringArguments rlt;
        bool bOK = false;
        if (!_is_empty_or_null(name))
        {
            arg.Put("Name", name);
        }
        return _Request("agent_svr_heart_beat", enum_ServerCall, 0, timeoutMs, &arg, NULL);
    }
    /// 查询客户端
    bool QueryClient(IStringArguments& rlt, uint timeoutMs)
    {
        return _Request("agent_svr_enum_client", enum_ServerCall, 0, timeoutMs, NULL, &rlt);
    }
    /// 查询第一个匹配的客户端
    uint QueryFirstClientId(const ByteArray& name, uint timeoutMs)
    {
        StringArguments rlt;
        if (!QueryClient(rlt, timeoutMs))
            return 0;
        uint count = rlt["Count"].To<uint>(0);
        list_t<string> ids;
        list_t<string> names;
        rlt.Get("ID", ids);
        rlt.Get("Name", names);
        list_t<string>::iterator id_itr = ids.begin();
        for (list_t<string>::iterator itr = names.begin(); itr != names.end(); ++itr, ++id_itr)
        {
            if (StringConvert::Contains(ByteArray(itr->c_str(), itr->length()), name))
                return ArgConvert::FromString<uint>(*id_itr);
        }
        return 0;
    }
    //----------------------------------------------------- 
};
//--------------------------------------------------------- 
class LocalAgent : public Runable, public IAgentNotifier, public RefObject
{
public:
    struct NotifyMessage
    {
        ByteBuilder cmd;
        StringArguments data;
        Timer::TickType expire_time;
    };
public:
    /// 通知过期时间
    uint NotifyTimeoutMs;
    /// 空闲等待间隔
    uint IdleIntervalMs;
protected:
    Ref<ICommandAgent> _api;

    std::list<NotifyMessage> _notify_pool;
public:
    LocalAgent() : Runable(), NotifyTimeoutMs(10000), IdleIntervalMs(100) {}

    virtual void Notify(const ByteArray& cmd, IStringArguments* arg)
    {
        _notify_pool.push_back(NotifyMessage());

        NotifyMessage& msg = _notify_pool.back();
        msg.cmd = cmd;
        msg.expire_time = Timer::TimeTick() + NotifyTimeoutMs;
        if (arg)
        {
            StringArguments::CopyTo(*arg, msg.data);
        }
    }
    virtual bool Update(Timer::TickType tick) { return _api->Update(tick); }
    virtual bool FixedUpdate(Timer::TickType tick, uint delta)
    {
        return _api->FixedUpdate(tick, delta);
    }

    void Serve(Ref<ICommandAgent> api)
    {
        _api = api;
        _api->observer = *this;
    }
    /// 发出请求
    bool Request(const ByteArray& cmd, IStringArguments* arg = NULL, IStringArguments* rlt = NULL)
    {
        Ref<ComplexCommand> command = _api->LookUp(cmd);
        if (command.IsNull())
        {
            if (rlt)
            {
                string msg = "<";
                msg += cmd.GetString();
                msg += "> is not exists";
                rlt->Put("Msg", msg);
            }
            return false;
        }
        StringArguments a;
        StringArguments r;
        return command->OnCommand(arg ? *arg : a, rlt ? *rlt : r);
    }
    /// 等待通知
    bool Expect(const ByteArray& notify, uint timeoutMs, IStringArguments* rlt = NULL)
    {
        Timer t;
        while (t.Elapsed() < timeoutMs)
        {
            for (std::list<NotifyMessage>::iterator itr = _notify_pool.begin(); itr != _notify_pool.end(); ++itr)
            {
                if (StringConvert::Compare(itr->cmd, notify))
                {
                    if (rlt)
                    {
                        StringArguments::CopyTo(itr->data, *rlt);
                    }
                    _notify_pool.erase(itr);
                    return true;
                }
            }
            Run(IdleIntervalMs);
        }
        return false;
    }
};
//--------------------------------------------------------- 
} // namespace driver
} // namespace application 
} // namespace zhou_yb
//--------------------------------------------------------- 
#endif // _LIBZHOUYB_COMMANDAGENT_H_
//=========================================================